#include "redis.hpp"

#include <iostream>
#include <thread>

using namespace std;

Redis::Redis() : publish_context(nullptr), subscribe_context(nullptr) {}
Redis::~Redis()
{
    if (publish_context != nullptr)
    {
        redisFree(publish_context);
    }

    if (subscribe_context != nullptr)
    {
        redisFree(subscribe_context);
    }
}

//连接redis服务器
bool Redis::connect()
{
    publish_context = redisConnect("127.0.0.1", 6379);
    if (publish_context == nullptr)
    {
        cerr << "connect redis failed!" << endl;
        return false;
    }

    subscribe_context = redisConnect("127.0.0.1", 6379);
    if (subscribe_context == nullptr)
    {
        cerr << "connect redis failed!" << endl;
        return false;
    }

    thread t([&]()
             { observer_channel_message(); });
    t.detach();

    cout << "connect redis-server success!" << endl;

    return true;
}
//发布消息
bool Redis::publish(int channel, string message)
{
    redisReply *reply = (redisReply *)redisCommand(publish_context, "PUBLISH %d %s", channel, message);
    if (reply == nullptr)
    {
        cerr << "publish command failed!" << endl;
        return false;
    }

    freeReplyObject(reply);

    return true;
}
//订阅频道
bool Redis::subscribe(int channel)
{
    if (REDIS_ERR == redisAppendCommand(this->subscribe_context, "SUBSCRIBE %d", channel))
    {
        cerr << "subscribe command failed" << endl;
        return false;
    }
    int done = 0;
    while (!done)
    {
        if (REDIS_ERR == redisBufferWrite(this->subscribe_context, &done))
        {
            cerr << "subscribe command failed!" << endl;
            return false;
        }
    }
    return true;
}

//取消订阅频道
bool Redis::unsubscribe(int channel) {
    if(REDIS_ERR == redisAppendCommand(this->subscribe_context, "UNSUBSCRIBE %d", channel)) {
        cerr << "unsubscribe command failed!" << endl;
        return false;
    }

    int done = 0;
    while(!done) {
        if(REDIS_ERR == redisBufferWrite(this->subscribe_context, &done)) {
           cerr << "unsubscribe command failed!" << endl; 
           return false;
        }
    }
    return true;
}
//接收频道消息
void Redis::observer_channel_message() {
    redisReply* reply = nullptr;
    while(REDIS_OK == redisGetReply(this->subscribe_context, (void **)&reply)) {
        if(reply != nullptr && reply->element[2] != nullptr && reply->element[2]->str != nullptr) {
            notify_message_handler(atoi(reply->element[1]->str), reply->element[2]->str);
        }

        freeReplyObject(reply);
    }

    cerr << "observer_channel_message quit" << endl;
}
//设置回调
void Redis::init_notify_handler(function<void(int, string)> fn) {
    this->notify_message_handler = fn;
}